package com.at.bigdata.spark.streaming

import org.apache.kafka.clients.consumer.ConsumerConfig
import org.apache.spark.SparkConf
import org.apache.spark.streaming.kafka010.{ConsumerStrategies, KafkaUtils, LocationStrategies}
import org.apache.spark.streaming.{Seconds, StreamingContext}

import java.io.{File, FileWriter, PrintWriter}
import java.text.SimpleDateFormat
import java.util.Date
import scala.collection.mutable.ListBuffer

/**
 *
 * @author cdhuangchao3
 * @date 2023/5/29 9:24 PM
 */
object SparkStreaming13_Req3 {

  def main(args: Array[String]): Unit = {
    // 创建环境
    // 创建时，需要传递2个参数：
    //    param： 环境配置
    val sc = new SparkConf().setMaster("local[*]").setAppName("operator")
    //    param2: 采集周期
    val ssc = new StreamingContext(sc, Seconds(5))

    val kafkaPara = Map[String, Object](
      ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG -> "linux1:9092,linux2:9092,linux3:9092",
      ConsumerConfig.GROUP_ID_CONFIG -> "at",
      ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG -> "org.apache.kafka.common.serialization.StringDeserializer",
      ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG -> "org.apache.kafka.common.serialization.StringDeserializer"
    )

    val kafkaDataDS = KafkaUtils.createDirectStream[String, String](
      ssc,
      LocationStrategies.PreferConsistent,
      ConsumerStrategies.Subscribe[String, String](Set("at"), kafkaPara)
    )

    val adClickData = kafkaDataDS.map(
      kafkaData => {
        val data = kafkaData.value()
        val datas = data.split(" ")
        AdClickData(datas(0), datas(1), datas(2), datas(3), datas(4))
      }
    )

    // 最近1分钟，每10秒计算一次
    // 12:01 => 12:00
    // 12:11 => 12:10
    // 12:19 => 12:10
    // 12:25 => 12:20
    // 12:59 => 12:50
    val reduceDS = adClickData.map(
      data => {
        val ts = data.ts.toLong
        val newTs = ts / 10000 * 10000
        (newTs, 1)
      }
    ).reduceByKeyAndWindow((x:Int,y:Int)=>x+y, Seconds(60), Seconds(10))
//    reduceDS.print()
    reduceDS.foreachRDD(
      rdd => {
        val list = ListBuffer[String]()
        val datas = rdd.sortByKey(true).collect()
        datas.foreach{
          case (time, cnt) => {
            val timeString = new SimpleDateFormat("mm:ss").format(new Date(time.toLong))
            list.append(s"""{"xtime":"${timeString}", "yval":"${cnt}"}""")
          }
        }
        // 输出文件
        val out = new PrintWriter(new FileWriter(new File("/Users/cdhuangchao3/project/code/spark_demo/spark_proj/adclick/adclick.json")))
        out.print("["+list.mkString(",")+"]")
        out.flush()
        out.close()
      }
    )

    // 1、启动采集器
    ssc.start()
    // 2、等待采集器的关闭
    ssc.awaitTermination()
  }

  // 广告点击数据
  case class AdClickData(ts: String, area: String, city: String, user: String, ad: String)
}
